package sparkStream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.Durations.seconds
import org.apache.spark.streaming.StreamingContext

object SparkStreaming {

  def main(args: Array[String]): Unit = {

    //  DStream   需要一个 spark_streming  jar 包
    //  1)  pom 文件 下载
    //   <dependency>
    //    <groupId>org.apache.spark</groupId>
    //    <artifactId>spark-streaming_2.11</artifactId>
    //    <version>2.3.0</version>
    //  </dependency>
    //  2)  jars 文件夹 ，包含 spark 需要的所有的jar包  （本地导入 jar 包 ）

    // 依然 加载配置项
    val conf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming")

    //   StreamingContext  加载 配置项  conf ， 设置时间间隔 1s
    val streamCon = new StreamingContext(conf,seconds(5))

    //  设置 log 的 level  级别；
    streamCon.sparkContext.setLogLevel("error")

    // 读取 流式数据  ；
    //      1) 读取 socket  端口数据；需要往 端口发送数据
    //      2) 处理 kafka （broker topic）的数据

    // 读取 socket  端口数据；需要往 端口发送数据
    val line =streamCon.socketTextStream("localhost",9998)

    // line  就是 DStream
    line.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()

    // 启动 streamCon ,  监控 的数据源
    streamCon.start()
    //  等待 监控 的数据源  停止
    streamCon.awaitTermination()






  }
}
